using System.Collections.Concurrent;
using System.Net;
using System.Net.Sockets;
using System.Text.Json;
using Microsoft.Extensions.Logging;
using SumerCoreDevOps.Infrastructure.Distributed;
using SumerCoreDevOps.Shared.Constants;
using SumerCoreDevOps.Shared.Enums;
using SumerCoreDevOps.Shared.Models;
using SumerCoreDevOps.Shared.Models.Messages;
using NetTcpClient = System.Net.Sockets.TcpClient;

namespace SumerCoreDevOps.Infrastructure.Tcp;

/// <summary>
/// TCP 服务器（控制面板端，支持多主节点消息转发）
/// </summary>
public class TcpServer : IDisposable
{
    private readonly ILogger<TcpServer> _logger;
    private readonly TcpMessageCodec _codec;
    private readonly ConcurrentDictionary<long, NetTcpClient> _connectedClients = new();
    private readonly IDistributedCoordinator? _coordinator;
    private readonly string? _masterId;
    private TcpListener? _listener;
    private CancellationTokenSource? _cts;
    private bool _isRunning;
    private bool _messageForwardingEnabled;

    public event Func<long, TcpMessageType, object, Task>? OnMessageReceived;
    public event Func<long, Task>? OnClientConnected;
    public event Func<long, Task>? OnClientDisconnected;

    public TcpServer(
        ILogger<TcpServer> logger,
        byte[] aesKey,
        IDistributedCoordinator? coordinator = null,
        string? masterId = null)
    {
        _logger = logger;
        _codec = new TcpMessageCodec(aesKey);
        _coordinator = coordinator;
        _masterId = masterId;
        _messageForwardingEnabled = coordinator != null && !string.IsNullOrEmpty(masterId);

        if (_messageForwardingEnabled)
        {
            _logger.LogInformation("TcpServer initialized with message forwarding enabled for master: {MasterId}", _masterId);
        }
    }

    /// <summary>
    /// 启动服务器
    /// </summary>
    public async Task StartAsync(int port, CancellationToken cancellationToken = default)
    {
        if (_isRunning)
        {
            throw new InvalidOperationException("Server is already running");
        }

        _listener = new TcpListener(IPAddress.Any, port);
        _listener.Start();
        _isRunning = true;
        _cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);

        _logger.LogInformation("TCP Server started on port {Port}", port);

        // 启动消息转发监听（如果启用）
        if (_messageForwardingEnabled && _coordinator != null)
        {
            await SubscribeToMessageForwardingAsync();
        }

        _ = Task.Run(async () => await AcceptClientsAsync(_cts.Token), _cts.Token);
    }

    /// <summary>
    /// 订阅消息转发频道
    /// </summary>
    private async Task SubscribeToMessageForwardingAsync()
    {
        if (_coordinator == null) return;

        try
        {
            await _coordinator.SubscribeAsync<MessageForwardRequest>(
                "channel:message_forward",
                async (request) =>
                {
                    try
                    {
                        // 检查是否有该节点的连接
                        if (_connectedClients.ContainsKey(request.TargetNodeId))
                        {
                            _logger.LogDebug("Forwarding message {MessageId} to node {NodeId} for master {SourceMasterId}",
                                request.MessageId, request.TargetNodeId, request.SourceMasterId);

                            await ForwardMessageToNodeAsync(request);
                        }
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError(ex, "Error forwarding message {MessageId} to node {NodeId}",
                            request.MessageId, request.TargetNodeId);
                    }
                });

            _logger.LogInformation("Subscribed to message forwarding channel");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to subscribe to message forwarding channel");
        }
    }

    /// <summary>
    /// 转发消息到节点
    /// </summary>
    private async Task ForwardMessageToNodeAsync(MessageForwardRequest request)
    {
        if (!_connectedClients.TryGetValue(request.TargetNodeId, out var client))
        {
            _logger.LogWarning("Cannot forward message {MessageId}: node {NodeId} not connected to this master",
                request.MessageId, request.TargetNodeId);
            return;
        }

        try
        {
            // 反序列化消息体（需要根据消息类型确定具体类型）
            var messageBody = JsonSerializer.Deserialize<object>(request.Payload);
            if (messageBody == null)
            {
                _logger.LogWarning("Cannot deserialize message payload for forwarding: {MessageId}", request.MessageId);
                return;
            }

            // 编码并发送
            byte[] data = _codec.Encode(request.MessageType, messageBody, false);
            NetworkStream stream = client.GetStream();
            await stream.WriteAsync(data);
            await stream.FlushAsync();

            _logger.LogInformation("Successfully forwarded message {MessageId} from {SourceMaster} to node {NodeId}",
                request.MessageId, request.SourceMasterId, request.TargetNodeId);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error forwarding message {MessageId} to node {NodeId}",
                request.MessageId, request.TargetNodeId);
        }
    }

    /// <summary>
    /// 停止服务器
    /// </summary>
    public Task StopAsync()
    {
        if (!_isRunning)
        {
            return Task.CompletedTask;
        }

        _isRunning = false;
        _cts?.Cancel();
        _listener?.Stop();

        // 关闭所有客户端连接
        foreach (var client in _connectedClients.Values)
        {
            client.Close();
            client.Dispose();
        }
        _connectedClients.Clear();

        _logger.LogInformation("TCP Server stopped");
        return Task.CompletedTask;
    }

    /// <summary>
    /// 发送消息给指定客户端（支持跨主节点转发）
    /// </summary>
    public async Task SendMessageAsync<T>(long nodeId, TcpMessageType messageType, T body, bool compress = false)
        where T : class
    {
        // 1. 优先尝试直接发送（节点连接到本主节点）
        if (_connectedClients.TryGetValue(nodeId, out var client))
        {
            try
            {
                byte[] data = _codec.Encode(messageType, body, compress);
                NetworkStream stream = client.GetStream();
                await stream.WriteAsync(data);
                await stream.FlushAsync();

                _logger.LogDebug("Sent message {MessageType} to node {NodeId}, size: {Size} bytes",
                    messageType, nodeId, data.Length);
                return;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to send message {MessageType} to node {NodeId}",
                    messageType, nodeId);
                throw;
            }
        }

        // 2. 如果节点未连接到本主节点，尝试通过消息转发
        if (_messageForwardingEnabled && _coordinator != null && _masterId != null)
        {
            try
            {
                _logger.LogDebug("Node {NodeId} not connected to this master, attempting message forwarding", nodeId);

                // 检查节点是否连接到其他主节点
                var nodeConnections = await _coordinator.HashGetAllAsync<DateTime>($"node:{nodeId}:connections");

                if (nodeConnections.Count == 0)
                {
                    throw new InvalidOperationException($"Node {nodeId} is not connected to any master");
                }

                // 创建转发请求
                var forwardRequest = new MessageForwardRequest
                {
                    TargetNodeId = nodeId,
                    MessageType = messageType,
                    Payload = JsonSerializer.Serialize(body),
                    SourceMasterId = _masterId
                };

                // 通过 Pub/Sub 发送转发请求
                var subscribers = await _coordinator.PublishAsync("channel:message_forward", forwardRequest);

                if (subscribers == 0)
                {
                    throw new InvalidOperationException($"No masters available to forward message to node {nodeId}");
                }

                _logger.LogInformation("Forwarded message {MessageType} for node {NodeId} via Pub/Sub, subscribers: {Subscribers}",
                    messageType, nodeId, subscribers);
                return;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to forward message {MessageType} to node {NodeId}",
                    messageType, nodeId);
                throw;
            }
        }

        // 3. 既未直接连接，也无法转发
        throw new InvalidOperationException($"Node {nodeId} not connected and message forwarding is disabled");
    }

    /// <summary>
    /// 广播消息给所有客户端
    /// </summary>
    public async Task BroadcastMessageAsync<T>(TcpMessageType messageType, T body, bool compress = false)
        where T : class
    {
        var tasks = _connectedClients.Keys.Select(nodeId =>
            SendMessageAsync(nodeId, messageType, body, compress));

        await Task.WhenAll(tasks);
    }

    /// <summary>
    /// 断开客户端连接
    /// </summary>
    private async Task DisconnectClientAsync(long nodeId)
    {
        if (_connectedClients.TryRemove(nodeId, out var client))
        {
            client.Close();
            client.Dispose();
            _logger.LogInformation("Client {NodeId} disconnected", nodeId);

            if (OnClientDisconnected != null)
            {
                await OnClientDisconnected(nodeId);
            }
        }
    }

    /// <summary>
    /// 接受客户端连接
    /// </summary>
    private async Task AcceptClientsAsync(CancellationToken cancellationToken)
    {
        while (_isRunning && !cancellationToken.IsCancellationRequested)
        {
            try
            {
                var client = await _listener!.AcceptTcpClientAsync(cancellationToken);

                // 生成临时节点ID（在注册时会替换为真实ID）
                long tempNodeId = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
                _connectedClients.TryAdd(tempNodeId, client);

                _logger.LogInformation("New client connected from {RemoteEndPoint}",
                    client.Client.RemoteEndPoint);

                // 启动消息接收循环
                _ = Task.Run(async () => await ReceiveMessagesAsync(tempNodeId, client, cancellationToken),
                    cancellationToken);
            }
            catch (OperationCanceledException)
            {
                break;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error accepting client connection");
            }
        }
    }

    /// <summary>
    /// 接收客户端消息
    /// </summary>
    private async Task ReceiveMessagesAsync(long nodeId, NetTcpClient client, CancellationToken cancellationToken)
    {
        NetworkStream stream = client.GetStream();
        byte[] headerBuffer = new byte[TcpProtocolConstants.MessageHeaderLength];

        try
        {
            while (_isRunning && !cancellationToken.IsCancellationRequested && client.Connected)
            {
                // 读取消息头
                int headerBytesRead = 0;
                while (headerBytesRead < TcpProtocolConstants.MessageHeaderLength)
                {
                    int read = await stream.ReadAsync(
                        headerBuffer.AsMemory(headerBytesRead,
                        TcpProtocolConstants.MessageHeaderLength - headerBytesRead),
                        cancellationToken);

                    if (read == 0)
                    {
                        // 客户端断开连接
                        await DisconnectClientAsync(nodeId);
                        return;
                    }

                    headerBytesRead += read;
                }

                // 解析消息头获取数据长度
                int dataLength = BitConverter.ToInt32(headerBuffer, 5);
                if (dataLength > TcpProtocolConstants.MaxMessageLength)
                {
                    _logger.LogWarning("Message too large from node {NodeId}: {Size} bytes",
                        nodeId, dataLength);
                    await DisconnectClientAsync(nodeId);
                    return;
                }

                // 读取完整消息（消息体 + CRC32）
                int totalLength = TcpProtocolConstants.MessageHeaderLength + dataLength + TcpProtocolConstants.CRC32Length;
                byte[] messageBuffer = new byte[totalLength];
                Array.Copy(headerBuffer, messageBuffer, TcpProtocolConstants.MessageHeaderLength);

                int bodyBytesRead = 0;
                int remainingBytes = dataLength + TcpProtocolConstants.CRC32Length;

                while (bodyBytesRead < remainingBytes)
                {
                    int read = await stream.ReadAsync(
                        messageBuffer.AsMemory(
                            TcpProtocolConstants.MessageHeaderLength + bodyBytesRead,
                            remainingBytes - bodyBytesRead),
                        cancellationToken);

                    if (read == 0)
                    {
                        await DisconnectClientAsync(nodeId);
                        return;
                    }

                    bodyBytesRead += read;
                }

                // 处理消息
                await ProcessMessageAsync(nodeId, messageBuffer);
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error receiving messages from node {NodeId}", nodeId);
            await DisconnectClientAsync(nodeId);
        }
    }

    /// <summary>
    /// 处理接收到的消息
    /// </summary>
    private async Task ProcessMessageAsync(long nodeId, byte[] messageData)
    {
        try
        {
            // 先解析消息头获取消息类型
            TcpMessageType messageType = (TcpMessageType)messageData[3];

            _logger.LogDebug("Received message {MessageType} from node {NodeId}", messageType, nodeId);

            // 根据消息类型解码
            object? body = messageType switch
            {
                TcpMessageType.NodeRegister => _codec.Decode<Shared.Models.Messages.NodeRegisterMessage>(messageData)?.Body,
                TcpMessageType.NodeAuth => _codec.Decode<Shared.Models.Messages.NodeAuthMessage>(messageData)?.Body,
                TcpMessageType.Heartbeat => _codec.Decode<Shared.Models.Messages.HeartbeatMessage>(messageData)?.Body,
                TcpMessageType.StepStatusReport => _codec.Decode<Shared.Models.Messages.StepStatusReportMessage>(messageData)?.Body,
                TcpMessageType.ContextSync => _codec.Decode<Shared.Models.Messages.ContextSyncMessage>(messageData)?.Body,
                TcpMessageType.CredentialRequest => _codec.Decode<Shared.Models.Messages.CredentialRequestMessage>(messageData)?.Body,
                TcpMessageType.LogStream => _codec.Decode<Shared.Models.Messages.LogStreamMessage>(messageData)?.Body,
                TcpMessageType.CleanupResult => _codec.Decode<Shared.Models.Messages.CleanupResultMessage>(messageData)?.Body,
                _ => null
            };

            if (body != null && OnMessageReceived != null)
            {
                await OnMessageReceived(nodeId, messageType, body);
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error processing message from node {NodeId}", nodeId);
        }
    }

    /// <summary>
    /// 更新客户端ID（注册后）
    /// </summary>
    public void UpdateClientId(long oldNodeId, long newNodeId)
    {
        if (_connectedClients.TryRemove(oldNodeId, out var client))
        {
            _connectedClients.TryAdd(newNodeId, client);
            _logger.LogInformation("Updated client ID from {OldId} to {NewId}", oldNodeId, newNodeId);
        }
    }

    /// <summary>
    /// 获取已连接的客户端数量
    /// </summary>
    public int GetConnectedClientsCount() => _connectedClients.Count;

    /// <summary>
    /// 检查客户端是否已连接
    /// </summary>
    public bool IsClientConnected(long nodeId) => _connectedClients.ContainsKey(nodeId);

    /// <summary>
    /// 获取所有已连接的节点ID
    /// </summary>
    public IEnumerable<long> GetConnectedNodeIds() => _connectedClients.Keys;

    public void Dispose()
    {
        StopAsync().Wait();
        _cts?.Dispose();
    }
}
